package com.lbjhan.rpc.consumer.server;

import com.lbjhan.rpc.common.RpcRequest;
import com.lbjhan.rpc.common.RpcResponse;
import com.lbjhan.rpc.consumer.handler.RpcNettyServerHandler;
import com.lbjhan.rpc.consumer.handler.RpcNettyServerRequestHandler;
import com.lbjhan.rpc.consumer.handler.RpcNettyServerSerializerDecoder;
import com.lbjhan.rpc.consumer.handler.RpcNettyServerSerializerEncoder;
import com.lbjhan.rpc.consumer.processor.ProviderBeanFactoryPostProcessor;
import com.lbjhan.rpc.handler.RpcCommonDeCoder;
import com.lbjhan.rpc.handler.RpcCommonEncoder;
import com.lbjhan.rpc.serializer.FastjsonSerializer;
import com.lbjhan.rpc.serializer.JacksonSerializer;
import com.lbjhan.rpc.serializer.KryoSerializer;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.ChannelFuture;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.codec.string.StringDecoder;
import io.netty.handler.codec.string.StringEncoder;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

/**
 * 基于Netty的服务端
 */
@Component
public class RpcNettyRequestServer implements DisposableBean {
    private NioEventLoopGroup bossGroup;
    private NioEventLoopGroup workerGroup;
    private CuratorFramework zk_server;

    @Autowired
    RpcNettyServerRequestHandler rpcNettyServerRequestHandler;

    public void startServer(String ip, int port){
        try {
            // 1. 创建线程组
            bossGroup = new NioEventLoopGroup(1);
            workerGroup = new NioEventLoopGroup();
            // 2. 创建服务端启动助手
            ServerBootstrap serverBootstrap = new ServerBootstrap();
            // 3. 设置参数
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childHandler(new ChannelInitializer<SocketChannel>() {

                        @Override
                        protected void initChannel(SocketChannel socketChannel) throws Exception {
                            ChannelPipeline pipeline = socketChannel.pipeline();
                            //添加序列化编解码器
//                            pipeline.addLast(new RpcNettyServerSerializerEncoder(new JacksonSerializer<RpcResponse>()));
//                            pipeline.addLast(new RpcNettyServerSerializerDecoder(new JacksonSerializer<RpcRequest>()));
                            pipeline.addLast(new RpcCommonEncoder(new KryoSerializer(), 1, 5));
                            pipeline.addLast(new RpcCommonDeCoder());
                            //业务处理类
                            pipeline.addLast(rpcNettyServerRequestHandler);
                        }
                    });
            // 4. 绑定端口号
            ChannelFuture channelFuture = serverBootstrap.bind(ip, port).sync();
            System.out.println("----- 服务器启动成功 -----");
            // 5. 注册服务
            register(ip, port);
            System.out.println("zookeeper注册成功");
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if(bossGroup != null)
                bossGroup.shutdownGracefully();

            if(workerGroup != null)
                workerGroup.shutdownGracefully();
        }

    }

    @Override
    public void destroy() throws Exception {
        if(bossGroup != null)
            bossGroup.shutdownGracefully();

        if(workerGroup != null)
            workerGroup.shutdownGracefully();
    }

    /**
     * 向zookeeper中注册服务
     * /rpc/serviceName/ip:port{data = cnt}
     */
    public void register(String ip, int port) throws Exception {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        zk_server = CuratorFrameworkFactory.builder()
                .connectString("101.200.199.134:2181")
                .sessionTimeoutMs(5000)
                .connectionTimeoutMs(3000)
                .retryPolicy(retryPolicy)
                .namespace("rpc")
                .build();
        zk_server.start();
        System.out.println("zookeeper 启动");
        String ip_port = "/"+ip+":"+port;
        //注册服务
        for(String serviceName : ProviderBeanFactoryPostProcessor.SERVICE_INSTANCE_MAP.keySet()){
            String path = "/" + serviceName + ip_port;
            zk_server.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    .forPath(path);
        }
    }
}
